package shardmaster

import (
	"fmt"
	"log"
	"sort"
	"sync"
	"time"

	"../labgob"
	"../labrpc"
	"../raft"
)

const debug = 0

//DPrintf :
func DPrintf(format string, a ...interface{}) (n int, err error) {
	if debug > 0 {
		log.Printf(format, a...)
	}
	return
}

//ShardMaster :
type ShardMaster struct {
	mu      sync.Mutex
	me      int
	rf      *raft.Raft
	applyCh chan raft.ApplyMsg

	// Your data here.

	configs []Config // indexed by config num

	//database   map[string]string
	reqHistory map[int64]int
	requestCh  chan Result
}

//Op :
type Op struct {
	// Your data here.
	ClientID int64
	ReqNum   int

	Operation string
	//Args      interface{}
	Servers map[int][]string
	GIDs    []int
	Shard   int
	GID     int
	Num     int
}

//Result :
type Result struct {
	WrongLeader bool
	Err         Err
	Config      Config
}

func (sm *ShardMaster) queryReqHistory(clientid int64, curNum int) bool {
	maxNum, ok := sm.reqHistory[clientid]
	if !ok {
		sm.reqHistory[clientid] = -1
		return false
	}
	return maxNum >= curNum
}

func (sm *ShardMaster) sendCommandAndReply(ok bool, command Op) bool {
	if !ok {
		term1, _, isleader := sm.rf.Start(command)
		if isleader {
			select {
			case <-sm.requestCh:
			case <-time.After(150 * time.Millisecond):
				return true
			}
			if term2, _ := sm.rf.GetState(); term2 != term1 {
				return true
			}
			return false
		}
		return true
	}
	return false

}

func (sm *ShardMaster) changeShard(newConfig *Config) {
	shardNum := 0
	var GIDs []int
	for k := range newConfig.Groups {
		GIDs = append(GIDs, k)
	}
	sort.Ints(GIDs)
	for shardNum < 10 {
		if len(newConfig.Groups) == 0 {
			newConfig.Shards[shardNum] = 0
			shardNum++
		}
		for _, k := range GIDs {
			newConfig.Shards[shardNum] = k
			shardNum++
			if shardNum >= 10 {
				break
			}
		}
	}
}

//Join :
func (sm *ShardMaster) Join(args *JoinArgs, reply *JoinReply) {
	// Your code here.
	sm.mu.Lock()
	ok := sm.queryReqHistory(args.ClientID, args.ReqNum)
	sm.mu.Unlock()
	command := Op{}
	command.ClientID = args.ClientID
	command.ReqNum = args.ReqNum
	command.Operation = "Join"
	command.Servers = args.Servers
	reply.WrongLeader = sm.sendCommandAndReply(ok, command)
}

//Leave :
func (sm *ShardMaster) Leave(args *LeaveArgs, reply *LeaveReply) {
	// Your code here.
	sm.mu.Lock()
	ok := sm.queryReqHistory(args.ClientID, args.ReqNum)
	sm.mu.Unlock()
	command := Op{}
	command.ClientID = args.ClientID
	command.ReqNum = args.ReqNum
	command.Operation = "Leave"
	command.GIDs = args.GIDs
	reply.WrongLeader = sm.sendCommandAndReply(ok, command)
}

//Move :
func (sm *ShardMaster) Move(args *MoveArgs, reply *MoveReply) {
	// Your code here.
	sm.mu.Lock()
	ok := sm.queryReqHistory(args.ClientID, args.ReqNum)
	sm.mu.Unlock()
	command := Op{}
	command.ClientID = args.ClientID
	command.ReqNum = args.ReqNum
	command.Operation = "Move"
	command.Shard = args.Shard
	command.GID = args.GID
	reply.WrongLeader = sm.sendCommandAndReply(ok, command)
}

//Query :
func (sm *ShardMaster) Query(args *QueryArgs, reply *QueryReply) {
	// Your code here.
	sm.mu.Lock()
	ok := sm.queryReqHistory(args.ClientID, args.ReqNum)
	sm.mu.Unlock()
	command := Op{}
	command.ClientID = args.ClientID
	command.ReqNum = args.ReqNum
	command.Operation = "Query"
	command.Num = args.Num
	if !ok {
		term1, _, isleader := sm.rf.Start(command)
		if isleader {
			var result Result
			select {
			case result = <-sm.requestCh:
			case <-time.After(150 * time.Millisecond):
				reply.WrongLeader = true
				return
			}
			if term2, _ := sm.rf.GetState(); term2 != term1 {
				reply.WrongLeader = true
				return
			}
			reply.Config = result.Config
			DPrintf("[%v]: query num %v, config %v", sm.me, args.Num, result.Config)
			reply.Err = OK
			reply.WrongLeader = false
		} else {
			reply.WrongLeader = true
		}
	} else {
		sm.mu.Lock()
		if args.Num == -1 || args.Num >= len(sm.configs) {
			reply.Config = sm.configs[len(sm.configs)-1]
		} else {
			reply.Config = sm.configs[args.Num]
		}
		sm.mu.Unlock()
		DPrintf("[%v]: old request query num %v, config %v", sm.me, args.ReqNum, reply.Config)
		reply.WrongLeader = false
		reply.Err = OK
	}
}

//applyRoutine :
func (sm *ShardMaster) applyRoutine() {
	for {
		applymsg := <-sm.applyCh
		if applymsg.CommandValid {
			sm.mu.Lock()
			if applymsg.CommandIndex == 0 {
				sm.mu.Unlock()
				continue
			}
			op, isOp := applymsg.Command.(Op)
			if ok := sm.queryReqHistory(op.ClientID, op.ReqNum); ok || !isOp {
				sm.mu.Unlock()
				continue
			}
			DPrintf("[%v]: command %v", sm.me, applymsg.Command)
			var result Result
			latestConfig := sm.configs[len(sm.configs)-1]
			newConfig := Config{}
			newConfig.Num = latestConfig.Num + 1
			newConfig.Groups = make(map[int][]string)
			for k, v := range latestConfig.Groups {
				newConfig.Groups[k] = v
			}
			switch op.Operation {
			case "Join":
				args := op
				for k, v := range args.Servers {
					_, ok := newConfig.Groups[k]
					if ok {
						newConfig.Groups[k] = append(newConfig.Groups[k], v...)
					} else {
						newConfig.Groups[k] = v
					}
				}
				//change shard
				//todo : need optimize
				sm.changeShard(&newConfig)
				sm.configs = append(sm.configs, newConfig)
				result.Err = OK
			case "Leave":
				args := op
				DPrintf("[%v]: leave args %v", sm.me, args)
				for _, key := range args.GIDs {
					delete(newConfig.Groups, key)
					DPrintf("[%v]: after delete key %v, newconfig groups %v", sm.me, key, newConfig.Groups)
				}
				//rechange the shard
				sm.changeShard(&newConfig)
				sm.configs = append(sm.configs, newConfig)
				DPrintf("[%v]: after leave, config: %v", sm.me, newConfig)
				result.Err = OK
			case "Move":
				args := op
				newConfig.Shards = latestConfig.Shards
				newConfig.Shards[args.Shard] = args.GID
				sm.configs = append(sm.configs, newConfig)
				result.Err = OK
			case "Query":
				args := op
				if args.Num == -1 || args.Num >= len(sm.configs) {
					result.Config = latestConfig
				} else {
					result.Config = sm.configs[args.Num]
				}
			default:
				fmt.Printf("Wrong Operation!\n")
			}
			DPrintf("[%v]: configs %v", sm.me, sm.configs)
			sm.reqHistory[op.ClientID] = op.ReqNum
			sm.mu.Unlock()
			select {
			case sm.requestCh <- result: //try to send channel to putappend or get RPC
			default:
			}
		} else {
			snapshotmsg := raft.ApplyMsg{}
			snapshotmsg.CommandValid = false
			sm.applyCh <- snapshotmsg
		}
	}
}

//Kill :
// the tester calls Kill() when a ShardMaster instance won't
// be needed again. you are not required to do anything
// in Kill(), but it might be convenient to (for example)
// turn off debug output from this instance.
//
func (sm *ShardMaster) Kill() {
	sm.rf.Kill()
	// Your code here, if desired.
}

//Raft :
// needed by shardkv tester
func (sm *ShardMaster) Raft() *raft.Raft {
	return sm.rf
}

//StartServer :
// servers[] contains the ports of the set of
// servers that will cooperate via Paxos to
// form the fault-tolerant shardmaster service.
// me is the index of the current server in servers[].
//
func StartServer(servers []*labrpc.ClientEnd, me int, persister *raft.Persister) *ShardMaster {
	sm := new(ShardMaster)
	sm.me = me

	sm.configs = make([]Config, 1)
	sm.configs[0].Groups = map[int][]string{}

	labgob.Register(Op{})
	sm.applyCh = make(chan raft.ApplyMsg)
	sm.rf = raft.Make(servers, me, persister, sm.applyCh)

	// Your code here.
	sm.reqHistory = make(map[int64]int)
	sm.requestCh = make(chan Result)
	go sm.applyRoutine()

	return sm
}
